package com.ruyuan.o2o.groupbuy.logcenter.hbase;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;


/**
 * hbase客户端
 *
 * @author ming qian
 */
@Component
@Slf4j
public class HBaseClient {

    public final static String HBASE_TABLE_NAME = "LOG";

    @Resource
    private HBaseConfig hBaseConfig;

    /**
     * 创建表
     *
     * @param serverName 前台传入的服务名
     * @param columns    字段
     */
    public boolean createTable(String serverName, String[] columns) throws IOException {
        Connection connection = hBaseConfig.getHbaseService();
        try (Admin admin = connection.getAdmin()) {
            TableName tableName = TableName.valueOf(serverName);
            if (admin.tableExists(tableName)) {
                log.info("Hbase表已存在");
                return true;
            }
            List<ColumnFamilyDescriptor> colFamilyList = new ArrayList<>();
            TableDescriptorBuilder tableDesBuilder = TableDescriptorBuilder.newBuilder(tableName);
            for (String str : columns) {
                ColumnFamilyDescriptor colFamilyDes = ColumnFamilyDescriptorBuilder.newBuilder(str.getBytes(StandardCharsets.UTF_8)).build();
                colFamilyList.add(colFamilyDes);
            }
            TableDescriptor tableDes = tableDesBuilder.setColumnFamilies(colFamilyList).build();
            admin.createTable(tableDes);
            log.info("Hbase建表完成：{}", serverName);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("Hbase建表失败");
            return false;
        }
    }

    /**
     * hbase 插入数据
     *
     * @param tableName  hbase表名
     * @param message    kafka message消息
     * @param documentId es index Id
     */
    public void insertHbase(String tableName, String message, String documentId) throws IOException {
        TableName tablename = TableName.valueOf(tableName);

        Put put = new Put(documentId.getBytes(StandardCharsets.UTF_8));
        //  保存参数：列族名、列名、值
        put.addColumn("logContent".getBytes(StandardCharsets.UTF_8),
                "logContent".getBytes(StandardCharsets.UTF_8),
                message.getBytes(StandardCharsets.UTF_8)
        );
        Connection connection = hBaseConfig.getHbaseService();
        Table table = connection.getTable(tablename);
        table.put(put);

        log.info("日志数据插入成功：documentId: {}", documentId);
    }

    /**
     * 单个查询hbase
     *
     * @param rowKey rowKey
     * @return 查询结果
     */
    public String queryTable(String rowKey) throws IOException {
        Connection connection = hBaseConfig.getHbaseService();
        Table table = connection.getTable(TableName.valueOf(HBASE_TABLE_NAME));

        Get get = new Get(rowKey.getBytes(StandardCharsets.UTF_8));
        Result result = table.get(get);
        if (result == null) {
            log.error("HBASE查询结果为空");
            return "";
        }
        //  返回查询结果
        Cell kv = result.rawCells()[0];
        return "null".equalsIgnoreCase(Bytes.toString(CellUtil.cloneValue(kv)))
                ? null
                : Bytes.toString(CellUtil.cloneValue(kv));
    }

}
